package com.zjc.springboot.storm;

import com.zjc.springboot.storm.bolt.KafkaBolt;
import com.zjc.springboot.storm.config.KafkaConfig;
import com.zjc.springboot.storm.spout.KafkaSpout;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.topology.TopologyBuilder;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.context.ConfigurableApplicationContext;

import java.util.Arrays;
import java.util.Properties;

@SpringBootApplication
public class StormMain implements CommandLineRunner {

    public static ConfigurableApplicationContext ac;

    private String connectString = "127.0.0.1:9092";
    private String topics = "TEST-STORM";
    private String group = "storm";

    private Config stormConf;

    private TopologyBuilder builder;

    public static void main(String[] args) {

        ac = new SpringApplicationBuilder(StormMain.class)
                .web(WebApplicationType.NONE)
                .run(args);
    }

    @Override
    public void run(String... args) throws Exception {

        KafkaConfig kafkaConfig = ac.getBean(KafkaConfig.class);
        this.builder = new TopologyBuilder();
        String stormMsgSpout = "stormMsgSpout";
        Properties kafkaPreperties = new Properties();
        kafkaPreperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, connectString);
        kafkaPreperties.put(ConsumerConfig.GROUP_ID_CONFIG, group);

        kafkaPreperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        kafkaPreperties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "3000");
        this.builder.setSpout(stormMsgSpout, new KafkaSpout(kafkaPreperties, Arrays.asList(topics.split(","))), 3);

        String stormMsgBolt = "stormMsgBolt";
        this.builder.setBolt(stormMsgBolt, new KafkaBolt(), 3).shuffleGrouping(stormMsgSpout);

        this.stormConf = new Config();

        this.stormConf.setDebug(false);

        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("toplogy", stormConf, builder.createTopology());
        Thread.sleep(Long.MAX_VALUE);
        cluster.shutdown();

    }
}
